package lib ;


import javax.annotation.Nullable;

import org.apache.flink.avro.shaded.org.apache.avro.Schema;
import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericData;
import org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;

import org.apache.flink.formats.avro.SchemaCoder;
import org.apache.flink.formats.avro.RegistryAvroDeserializationSchema;
import org.apache.flink.formats.avro.registry.confluent.ConfluentSchemaRegistryCoder;



public class ExtendedConfluentRegistryAvroDeserializationSchema<T> extends RegistryAvroDeserializationSchema<T> {
    private static final int DEFAULT_IDENTITY_MAP_CAPACITY = 1000;
    private static final long serialVersionUID = -1671641202177852775L;

    private ExtendedConfluentRegistryAvroDeserializationSchema(Class<T> recordClazz,  @Nullable Schema reader, SchemaCoder.SchemaCoderProvider schemaCoderProvider) {
        super(recordClazz, reader, schemaCoderProvider);
    }

    public static ExtendedConfluentRegistryAvroDeserializationSchema<GenericData.Array> forGenericArray(Schema schema, String url) {
        return forGenericArray(schema, url, DEFAULT_IDENTITY_MAP_CAPACITY);
    }
    public static ExtendedConfluentRegistryAvroDeserializationSchema<GenericData.Array> forGenericArray(Schema schema, String url, int identityMapCapacity) {
        return new ExtendedConfluentRegistryAvroDeserializationSchema<GenericData.Array>(GenericData.Array.class, schema, new CachedSchemaCoderProvider(url, identityMapCapacity));
    }

    private static class CachedSchemaCoderProvider implements SchemaCoder.SchemaCoderProvider {
        private static final long serialVersionUID = 4023134423033312666L;
        private final String url;
        private final int identityMapCapacity;

        CachedSchemaCoderProvider(String url, int identityMapCapacity) {
            this.url = url;
            this.identityMapCapacity = identityMapCapacity;
        }

        @Override
        public SchemaCoder get() {
            return new ConfluentSchemaRegistryCoder(new CachedSchemaRegistryClient(url, identityMapCapacity));
        }
    }
}
